#ifndef _M__BROKER__H_
#define _M__BROKER__H_
#include "../thirdLib/proto/codec.h"
#include "../thirdLib/proto/dispatcher.h"
#include "../thirdLib/base/Mutex.h"
#include "../thirdLib/net/EventLoop.h"
#include "../thirdLib/net/TcpServer.h"
#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "mq_virtual_host.hpp"
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_thread_pool.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include "../mqCommon/mq_proto.pb.h"

/*

*/
namespace yqx
{
    #define DBFILE "/meta.db"
    class BrokerServer
    {
    public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
        BrokerServer(int port, const std::string& basedir)
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
              _dispatcher(std::bind(&BrokerServer::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
              _pcodec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _vhp(std::make_shared<VirtualHost>(basedir, basedir + DBFILE)),
              _cmp(std::make_shared<ConsumerManager>()),
              _cnp(std::make_shared<ConnectionManager>()),
              _tpp(std::make_shared<ThreadPool>())
        {
            //针对历史消息中的所有队列，初始化队列的消费者管理结构
            QueueMap qm = _vhp->getAllQueues();
            for(auto& q : qm)
                _cmp->initQueueConsumer(q.first);
            //注册处理请求的回调函数
            _dispatcher.registerMessageCallback<yqx::openChannelRequest>(std::bind(&BrokerServer::onOpenChannal, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannal, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::queueBindRequest>(std::bind(&BrokerServer::onQueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::queueUnbindRequest>(std::bind(&BrokerServer::onQueueUnbind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::basicAckRequest>(std::bind(&BrokerServer::onBasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::basicConsumeRequest>(std::bind(&BrokerServer::onBasicConsume, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<yqx::basicCancelRequest>(std::bind(&BrokerServer::onBasicConcle, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _pcodec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setConnectionCallback(std::bind(&BrokerServer::onConnection, this, std::placeholders::_1));
        }
        void start()
        {
            _server.start();
            _baseloop.loop();
        }

    private:
        //打开信道
        void onOpenChannal(const::muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("打开信道时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            return mcoon->openChannal(message);
        }
        //删除信道
        void onCloseChannal(const::muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("关闭信道时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            return mcoon->closeChannal(message);
        }
        //声明交换机
        void onDeclareExchange(const::muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("声明交换机时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("声明交换机时,未找到信道!");
                return;
            }
            return cp->declareExchange(message);
        }
        //删除交换机
        void onDeleteExchange(const::muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("删除交换机时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("删除交换机时,未找到信道!");
                return;
            }
            return cp->deleteExchange(message);
        }
        //声明队列
        void onDeclareQueue(const::muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("声明队列时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("声明队列时,未找到信道!");
                return;
            }
            return cp->declareQueue(message);
        }
        //删除队列
        void onDeleteQueue(const::muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("删除队列时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("删除队列时,未找到信道!");
                return;
            }
            return cp->deleteQueue(message);
        }
        //队列绑定
        void onQueueBind(const::muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("队列绑定时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("队列绑定时,未找到信道!");
                return;
            }
            return cp->bind(message);
        }
        //队列解绑
        void onQueueUnbind(const::muduo::net::TcpConnectionPtr& conn, const queueUnbindRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("队列解绑时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("队列解绑时,未找到信道!");
                return;
            }
            return cp->unbind(message);
        }
        //消息发布
        void onBasicPublish(const::muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("发布消息时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("发布消息时,未找到信道!");
                return;
            }
            return cp->basicPublish(message);
        }
        //消息确认
        void onBasicAck(const::muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("发布确认时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("发布确认时,未找到信道!");
                return;
            }
            return cp->basicAck(message);
        }
        //队列消息订阅
        void onBasicConsume(const::muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("订阅队列消息时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("订阅队列消息时,未找到信道!");
                return;
            }
            return cp->basicConsume(message);
        }
        //队列消息取消订阅
        void onBasicConcle(const::muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message, muduo::Timestamp){
            Connection::ptr mcoon = _cnp->getConnection(conn);
            if(mcoon == nullptr){
                LOG("取消订阅时,未找到连接对应的connection对象!");
                conn->shutdown();
                return;
            }
            Channal::ptr cp = mcoon->getChannal(message->cid());
            if(cp == nullptr){
                LOG("取消订阅时,未找到信道!");
                return;
            }
            return cp->basicCancle(message);
        }
        //未知消息
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            std::cout << "onUnknownMessage: " << message->GetTypeName() << std::endl;
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
                _cnp->createConnection(conn, _pcodec, _cmp, _vhp, _tpp);
            else
                _cnp->deleteConnection(conn);
        }       

    private:
        muduo::net::EventLoop _baseloop;// 事件监控对象
        muduo::net::TcpServer _server;  // 服务器对象
        ProtobufDispatcher _dispatcher; // 请求分法器对象--要向其中注册请求处理函数
        ProtobufCodecPtr _pcodec;       // protobuf协议处理器--对收到的数据进行protobuf协议处理
        VirtualHost::ptr _vhp;
        ConsumerManager::ptr _cmp;
        ConnectionManager::ptr _cnp;
        ThreadPool::ptr _tpp;
    };
}
#endif